/**
 * 消息传输
 *
 * Created by snowingsea on 2020/02/02.
 */
const url = require('url');
const { v4: uuidv4 } = require('uuid');
const auth = require('./auth');
const users = require('./users');
const heartbeat = require('./heartbeat');
const utility = require('./utility');
const publisher = require('./redis/publisher');
// const redis = require('./redis');
const PhoenixError = require('./phoenix-error');
const MessageSubscriber = require('./amqp/pub-sub/subscriber');
const logger = require('./logger');

/**
 * websocket关闭状态码
 * @type {Object}
 */
const closeCode = {
  normal: 1000,
  goingAway: 1001,
  protocolError: 1002,
  unsupported: 1003,
  noStatus: 1005,
  abnormal: 1006,
  unsupportedPayload: 1007,
  policyViolation: 1008,
  tooLarge: 1009,
  serverError: 1011,
  serviceRestart: 1012,
  tryAgainLater: 1013,
  badGateway: 1014,
  tlsHandshakeFail: 1015,
  forceClose: 4001,
  heartbeatTimeout: 4002,
};

/**
 * 发送消息
 * @param {Object} ws webService对象
 * @param {Object|String} msg 消息内容
 */
function sendMessage(ws, msg) {
  if (utility.isObjectType(msg)) {
    msg = JSON.stringify(msg);
  }
  try {
    ws.send(msg);
  } catch (e) {
    ws.close(closeCode.serverError, `Fail to send message: ${msg}`);
  }
}

/**
 * 消息通信
 * @param {Object} ws webService对象
 * @param {MessageSubscriber} subClient 订阅客户端
 * @param {String} accessToken 当前用户访问token
 */
function communicate(ws, subClient, accessToken) {
  subClient.on('message', utility.actionMerge((channel, msg) => {
    const message = JSON.parse(msg);

    if (message.cmd === 'logout') {
      if (!message.data.currentAccessToken || message.data.currentAccessToken !== accessToken) {
        ws.close(closeCode.forceClose, 'Force to logout');
      }
      return;
    }
    sendMessage(ws, msg);
  }, {
    mergeFilter: (channel, msg) => {
      const message = JSON.parse(msg);
      // 文件上传、报告下载等进度消息 直接发送，不合并
      // note: 此消息请前端仅仅作为进度展示，不要用于任何逻辑判断
      return message.domain !== 'progress';
    },
    argReduce: (oldArgs, args) => {
      const [channel] = args;
      let [, data] = args;
      let [, dataOld] = oldArgs;
      data = JSON.parse(data);
      dataOld = JSON.parse(dataOld);
      switch (data.cmd) {
        case 'add':
        case 'update':
        {
          if (dataOld.cmd === 'add') {
            data.cmd = 'add';
          }
          return [channel, JSON.stringify(utility.deepMerge(dataOld, data))];
        }

        default:
          throw new Error('only merge update message');
      }
    },
    namespace: ([channel, msg]) => {
      const { cmd, data, domain } = JSON.parse(msg);
      if (cmd === 'update' || cmd === 'add') {
        // 把update合并到add里去
        const cmd1 = 'add_or_update';
        switch (domain) {
          case 'project':
            return `${channel}_${cmd1}_${domain}_${data.projectId}`;
          case 'projectVersion':
            return `${channel}_${cmd1}_${domain}_${data.versionId}`;
          case 'file':
            return `${channel}_${cmd1}_${domain}_${data.fileId}`;
          case 'function':
            return `${channel}_${cmd1}_${domain}_${data.functionId}`;
          case 'testCase':
            return `${channel}_${cmd1}_${domain}_${data.testId}`;
          case 'task':
            return `${channel}_${cmd1}_${domain}_${data.taskId}`;
          case 'codeDefect':
            return `${channel}_${cmd1}_${domain}_${data.functionId}_${data.defectId}`;
          case 'hash':
            return data.creatorId.toString();
          default:
            return undefined;
        }
      }
      return undefined;
    },
    debounceMs: 1000,
    timeoutMs: 5000,
  }));
}

/**
 * WebService连接
 * @param {Object} ws webService对象
 * @param {Object} req 请求参数
 */
function connect(ws, req) {
  const { accessToken } = url.parse(req.url, true).query;
  let subClient;
  let user;
  const uuid = uuidv4();
  auth.getToken(accessToken)
    .then(({ userId }) => users.getUser(userId))
    .then((res) => {
      user = res;
      const userId = user._id; // eslint-disable-line no-underscore-dangle
      switch (user.status) {
        case users.statusKind.deleting:
          return Promise.reject(new PhoenixError('not found', 'The user is not exists'));
        case users.statusKind.online:
          // 踢出在线用户
          publisher.publishLogoutMessage(userId, accessToken);
          return Promise.resolve();
        default:
          return users.updateUser(userId, { status: users.statusKind.online });
      }
    })
    .then(async () => {
      const ms = new MessageSubscriber('ex_websocket');
      await ms.initQueue();
      return ms;
    })
    .then(async (client) => {
      await heartbeat.keepAlive(uuid, user._id, true);
      subClient = client;
      ws.on('message', async (data) => {
        // TODO: 权限控制
        const message = JSON.parse(data);
        switch (message.cmd) {
          case 'subscribe':
            subClient.subscribe(message.data.channel);
            break;
          case 'unsubscribe':
            subClient.unsubscribe(message.data.channel);
            break;
          case 'heartbeat':
            {
              // 每次都检查一遍token是否依然有效
              const tokenExisted = await auth.getToken(accessToken);
              if (!tokenExisted) {
                // 当前token已失效; 可能是过期，也有可能因为人数上限而被踢
                sendMessage(ws, { cmd: 'refuse' });
                ws.close(closeCode.forceClose, 'Token invalid');
              } else {
                const { n } = await heartbeat.keepAlive(uuid, user._id, false);
                if (n === 0) {
                  // 心跳已超时
                  sendMessage(ws, { cmd: 'refuse' });
                  ws.close(closeCode.heartbeatTimeout, 'Heartbeat timeout');
                }
              }
            }
            break;

          default:
            break;
        }
      });
      sendMessage(ws, { cmd: 'accept' });
      const userId = user._id; // eslint-disable-line no-underscore-dangle
      // 订阅私人频道
      await subClient.subscribe(publisher.privateChannel(userId));
      // 用户状态更新
      publisher.publishUserMessage('update', { userId, status: users.statusKind.online });
      communicate(ws, subClient, accessToken);

      ws.on('close', async () => {
        await heartbeat.disconnect(uuid);
        const isOnline = await heartbeat.isUserOnline(userId);
        if (!isOnline) {
          users.updateUser(userId, { status: users.statusKind.offline })
            .then(() => {})
            .catch(() => {});
          // 用户状态更新
          publisher.publishUserMessage('update', { userId, status: users.statusKind.offline });
        }
        if (subClient) {
          subClient.end(true);
        }
      });
    })
    .catch((err) => {
      sendMessage(ws, { cmd: 'refuse' });
      logger.error('', err);
      ws.close(closeCode.serverError, 'Fail to connect websocket: server error');
    });
}

/**
 * Export model definition object.
 */
module.exports = {
  closeCode,
  connect,
};
